package com.shujia.spark.core

import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

import scala.io.Source

object Demo20Bro {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()

    conf.setAppName("acc")

    conf.setMaster("local")

    val sc = new SparkContext(conf)

    /**
     * 广播变量： 当在算子内使用算子外的一个大变量是可以将这个变量广播出去，可以减少变量副本数
     *
     */
    val scoreRDD: RDD[String] = sc.textFile("data/score.txt")


    //读取学生便得到一个集合
    val studentList: List[String] = Source.fromFile("data/students.txt").getLines().toList

    //将学号表转换成map集合
    val studentMap: Map[String, String] = studentList
      .map((stu: String) => {
        val split: Array[String] = stu.split(",")
        val id: String = split(0)
        (id, stu)
      })
      .toMap

    //1、在Driver端将一个普通变量广播出去
    val stuBro: Broadcast[Map[String, String]] = sc.broadcast(studentMap)

    //使用map集合进行表管理
    val joinRDD: RDD[(String, String)] = scoreRDD.map((sco: String) => {

      val split: Array[String] = sco.split(",")
      val id: String = split(0)

      /**
       * 当在算子内使用算子外的一个变量时，这个变量会作为变量副本封装到task中， 发生到Executor中执行，
       * 变量副本的数据等于task的数量，task的数量等于分区的数据
       * 副本太多会导致整体效率降低
       *
       * 使用广播变量
       * 1、将变量广播到Executor中，task不需要自己带一个副本过来，每一个Executor中指挥保存一个变量副本
       * 2、如果不适用广播变量，每一个task中一个变量的副本
       *
       */

      //2、在Executor端获取广播变量
      val student: Map[String, String] = stuBro.value

      //使用id冲学生表额map集合中获取学生信息
      val stuInfo: String = student.getOrElse(id, "默认值")

      (sco, stuInfo)
    })

    joinRDD.foreach(println)


  }

}
